package com.zmn.oms.task.es.business.impl.ordercomment;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.zmn.common.utils.date.DateUtil;
import com.zmn.common.utils.string.StringUtil;
import com.zmn.manager.mq.constant.MqTopicConsts;
import com.zmn.manager.mq.interfaces.ZmnMQSender;
import com.zmn.oms.manager.utils.OmsMqTagConsts;
import com.zmn.oms.services.interfaces.work.OrderWorkService;
import com.zmn.oms.task.es.business.interfaces.ordercomment.RebuildEsOrderCommentService;
import com.zmn.oms.task.es.business.interfaces.ordercomment.SyncEsOrderCommentBService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;

/**
 * 类描述：订单评价重建es索引
 *
 * @author lujia
 * @date 2018/11/23 19:48
 */
@Service
public class RebuildEsOrderCommentServiceImpl implements RebuildEsOrderCommentService {

    private static final Logger logger = LoggerFactory.getLogger(RebuildEsOrderCommentServiceImpl.class);
    private static final String SQL_SELECT = "SELECT order_id FROM `oms_order_comment_extend` WHERE create_time >= '%s' AND create_time < '%s'";
    private static int DEFAULT_PAGE_SIZE = 100;
    @Autowired
    SyncEsOrderCommentBService syncEsOrderCommentBService;
    @Autowired
    OrderWorkService orderWorkService;

    @Autowired
    ZmnMQSender zmnMQSender;
    @Autowired
    @Qualifier("masterDataSource")
    DataSource dataSource;

    @Override
    public void rebuild(Date startTime, Date endTime, Integer number) throws SQLException {
        int pageSize = number == null ? DEFAULT_PAGE_SIZE : number;
        String startTimeStr = DateUtil.dateFormatToString(startTime, DateUtil.FORMAT_DEFAULT);
        String endTimeStr = DateUtil.dateFormatToString(endTime, DateUtil.FORMAT_DEFAULT);

        logger.debug("#es#rebuild#ordercomment -> startTime : {}, endTime : {}, number : {}", startTimeStr, endTimeStr, number);

        processTableData(startTimeStr, endTimeStr, pageSize);
    }

    /**
     * 处理单表数据
     *
     * @param startTimeStr
     * @param endTimeStr
     * @param pageSize
     * @throws SQLException
     */
    private void processTableData(String startTimeStr, String endTimeStr, int pageSize) throws SQLException {

        Connection connection = dataSource.getConnection();

        String sql = String.format(SQL_SELECT, startTimeStr, endTimeStr);
        // 设置游标只能向下移动 防止 oom
        PreparedStatement stmt = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);

        // 一次加载到内存的数量
        stmt.setFetchSize(Integer.MIN_VALUE);
        ResultSet resultSet = stmt.executeQuery();

        int count = 0;
        List<Long> orderIds = Lists.newArrayListWithCapacity(pageSize);

        while (resultSet.next()) {
            long orderId = resultSet.getLong("order_id");
            orderIds.add(orderId);
            count++;
            if (count % pageSize == 0) {
                sendMq(orderIds);
                orderIds.clear();
            }
        }

        if (!orderIds.isEmpty()) {
            sendMq(orderIds);
        }

        logger.debug("#es#rebuild#ordercomment#processTableData num : {}", count);
        connection.close();
    }

    private void sendMq(List<Long> orderIds) {
        String key = StringUtil.getRandomsString(32);
        logger.debug("#es#rebuild#ordercomment#sendMq key : {}", key);
        zmnMQSender.send(MqTopicConsts.ZMN_TOPIC_OMS, OmsMqTagConsts.ZMN_TOPIC_OMS_CHANGE_ORDER_COMMENT, key, JSON.toJSONString(orderIds));
    }
}
